package com.captjack.bigdata.storm.topology;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * Bolt是一个使用元组作为输入，处理元组，并产生新的元组作为输出的组件
 *
 * @author Jack Sparrow
 * @version 1.0.0
 * @date 2018/4/13 00:22
 * package com.cc.bigdata.storm.topology
 */
public class StreamDataBolt implements IRichBolt {

    /**
     * outputCollector
     */
    private OutputCollector outputCollector;

    /**
     * 为bolt提供要执行的环境。执行器将运行此方法来初始化spout
     *
     * @param map             为此bolt提供Storm配置
     * @param topologyContext 提供有关拓扑中的bolt位置，其任务ID，输入和输出信息等的完整信息
     * @param outputCollector 发出处理的元组
     */
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector = outputCollector;
    }

    /**
     * execute方法一次处理单个元组。元组数据可以通过Tuple类的getValue方法访问。
     * 不必立即处理输入元组。多元组可以被处理和输出为单个输出元组。处理的元组可
     * 以通过使用OutputCollector类发出
     *
     * @param tuple 元组
     */
    @Override
    public void execute(Tuple tuple) {
        String from = tuple.getString(0);
        String to = tuple.getString(1);
        int duration = tuple.getInteger(2);
        outputCollector.emit(new Values(from + " - " + to, duration));
    }

    @Override
    public void cleanup() {

    }

    /**
     * 这里的参数declarer用于声明输出流id，输出字段等。
     * 此方法用于指定元组的输出模式
     *
     * @param outputFieldsDeclarer 元组输出格式
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("call", "duration"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}
